package com.hisense.ovcloud.spark.sql.service;

import cn.hutool.json.JSONUtil;
import com.hisense.ovcloud.spark.sql.config.Conf;
import com.hisense.ovcloud.spark.sql.dto.CustomDevicePairDto;
import com.hisense.ovcloud.spark.sql.dto.DeviceNicknameKafkaMsg;
import com.hisense.ovcloud.spark.sql.entity.CustomDevicePair;
import com.hisense.ovcloud.spark.sql.entity.DeviceSupportFeatureCode;
import com.hisense.ovcloud.spark.sql.entity.DeviceSupportPuid;
import com.mybatisflex.core.row.Db;
import io.micrometer.common.util.StringUtils;
import io.smallrye.mutiny.tuples.Tuple2;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;

import static com.hisense.ovcloud.spark.sql.config.Conf.PAIR_END_TIME_SLOT;

@Service
@Slf4j
public class ReactiveConsumerService implements CommandLineRunner, InitializingBean {

    @Value("${topics.deviceNickNameChange:deviceNickNameChange}")
    private String deviceNickNameChange;

    @Value("${topics.modelConfigParam:model_config_param}")
    private String modelConfigParam;

    @Value("${topics.devicePowerFunctionMsg:device_power_function_msg}")
    private String devicePowerFunctionMsg;

    @Value("${ignoreIds:}")
    private String ignorePUIds;

    @Value("${poolSize:20}")
    private int poolSize;

    @Value("${waitQueueLength:1000}")
    private int waitQueueLength;

    private final static String DEFAULT_ZERO_TIMEZONE_ID = "Etc/GMT0";

    private static Scheduler workerPool;

    final
    ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate;

    public ReactiveConsumerService(ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate) {
        this.kafkaConsumerTemplate = kafkaConsumerTemplate;
    }

    private Flux<Mono<Boolean>> msgConsummer() {
        return kafkaConsumerTemplate
                .receiveAutoAck()
                .map(this::handler);
    }

    private final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    protected Mono<Boolean> handler(ConsumerRecord<String, String> consumerRecord) {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {
            if (StringUtils.isNotEmpty(consumerRecord.value())) {
                String receivedTxt = consumerRecord.value().strip().replaceAll("\n", "");
                log.info("receive record: {}", receivedTxt);
                boolean flag = !StringUtils.isNotEmpty(ignorePUIds) || Arrays.stream(ignorePUIds.split(",")).noneMatch(receivedTxt::contains);
                if(flag){
                    Mono.just(Tuple2.of(receivedTxt, consumerRecord)).publishOn(workerPool).subscribeOn(workerPool).subscribe(tuple -> disposeMsg(tuple.getItem1(), tuple.getItem2()));
                }
            }
        } catch (Exception e) {
            log.error("consuming encountered error: {}", e.getMessage());
        }
        stopWatch.stop();
        log.info("This batch consume total {} ms", stopWatch.getTotalTimeMillis());
        return Mono.just(true);
    }


    private void disposeMsg(String receivedTxt, ConsumerRecord<String, String> consumerRecord){
        if (consumerRecord.topic().equals(deviceNickNameChange)) {
            DeviceNicknameKafkaMsg deviceNickNameMsg = JSONUtil.toBean(receivedTxt, DeviceNicknameKafkaMsg.class);
            log.info("try to update nickname {} to a row customid {}, puid {}, platformId {}", deviceNickNameMsg.getDevicenikename(), deviceNickNameMsg.getCustomerid(), deviceNickNameMsg.getPuid(), deviceNickNameMsg.getPlatformid());
            CustomDevicePair.model()
                    .setPairAlias(deviceNickNameMsg.getDevicenikename())
                    .where(CustomDevicePair::getCustomerId).eq(deviceNickNameMsg.getCustomerid())
                    .where(CustomDevicePair::getPuid).eq(deviceNickNameMsg.getPuid())
                    .where(CustomDevicePair::getPlatform).eq(deviceNickNameMsg.getPlatformid())
                    .where(CustomDevicePair::getPairEnd).eq(PAIR_END_TIME_SLOT)
                    .update();
        } else if (consumerRecord.topic().equals(modelConfigParam)) {
            DeviceSupportFeatureCode deviceSupportFeatureCode = JSONUtil.toBean(receivedTxt, DeviceSupportFeatureCode.class);
            log.info("try to insert DeviceSupportFeatureCode: featurecode {},", receivedTxt);
            Db.insertBySql(Conf.DeviceSupportFeatureCode,
                    deviceSupportFeatureCode.getDevicetype(),
                    deviceSupportFeatureCode.getFeaturecode(),
                    deviceSupportFeatureCode.getPower(),
                    deviceSupportFeatureCode.getWator(),
                    deviceSupportFeatureCode.getOrigin());

        } else if (consumerRecord.topic().equals(devicePowerFunctionMsg)) {
            DeviceSupportPuid deviceSupportPuid = JSONUtil.toBean(receivedTxt, DeviceSupportPuid.class);
            log.info("try to insert deviceSupportPuid: puid {},", receivedTxt);
            Db.insertBySql(Conf.DeviceSupportPuid,
                    deviceSupportPuid.getPuid(),
                    deviceSupportPuid.getDevicetype(),
                    deviceSupportPuid.getFeaturecode(),
                    deviceSupportPuid.getPower(),
                    deviceSupportPuid.getWator()
                    );
        } else {
            CustomDevicePairDto customDevicePairMsg = JSONUtil.toBean(receivedTxt, CustomDevicePairDto.class);
            TimeZoneCalculateExt timeZoneCalculate = calculateZoneDateTime(customDevicePairMsg.getTimezone(), customDevicePairMsg.getTime());

            if (customDevicePairMsg.getActiontype().equals(Conf.INSERT_ACTION_TYPE)) {
                log.info("try to insert a row customid {}, puid {}, platformId {}", customDevicePairMsg.getCustomerid(), customDevicePairMsg.getPuid(), customDevicePairMsg.getPlatformid());
                if(timeZoneCalculate.isFlag()){
                    CustomDevicePair.model()
                            .setCustomerId(customDevicePairMsg.getCustomerid())
                            .setPuid(customDevicePairMsg.getPuid())
                            .setWifiId(customDevicePairMsg.getWifiid())
                            .setDeviceId(customDevicePairMsg.getDeviceid())
                            .setDeviceType(customDevicePairMsg.getDevicetype())
                            .setPairAlias(customDevicePairMsg.getDevicenikename())
                            .setFeatureCode(customDevicePairMsg.getFeaturecode())
                            .setPlatform(customDevicePairMsg.getPlatformid())
                            .setPairBegin(timeZoneCalculate.getTimeZoneCalculate().getDatetime())
                            .setTimezone(timeZoneCalculate.getTimeZoneCalculate().getTimezone())
                            .setTimezoneId(customDevicePairMsg.getTimezone())
                            .saveOpt()
                            .orElseThrow();
                }else{
                    CustomDevicePair.model()
                            .setCustomerId(customDevicePairMsg.getCustomerid())
                            .setPuid(customDevicePairMsg.getPuid())
                            .setWifiId(customDevicePairMsg.getWifiid())
                            .setDeviceId(customDevicePairMsg.getDeviceid())
                            .setDeviceType(customDevicePairMsg.getDevicetype())
                            .setPairAlias(customDevicePairMsg.getDevicenikename())
                            .setFeatureCode(customDevicePairMsg.getFeaturecode())
                            .setPlatform(customDevicePairMsg.getPlatformid())
                            .setPairBegin(timeZoneCalculate.getTimeZoneCalculate().getDatetime())
                            .setTimezone(timeZoneCalculate.getTimeZoneCalculate().getTimezone())
                            .setTimezoneId(DEFAULT_ZERO_TIMEZONE_ID)
                            .saveOpt()
                            .orElseThrow();
                }

            } else {
                List<CustomDevicePair> queryList = CustomDevicePair.model()
                        .where(CustomDevicePair::getCustomerId).eq(customDevicePairMsg.getCustomerid())
                        .where(CustomDevicePair::getPuid).eq(customDevicePairMsg.getPuid())
                        .where(CustomDevicePair::getPlatform).eq(customDevicePairMsg.getPlatformid())
                        .where(CustomDevicePair::getPairEnd).eq(PAIR_END_TIME_SLOT)
                        .list();

                if (!queryList.isEmpty()) {
                    log.info("will update pair end time {} for a row with customid {}, puid {}, platformId {}", timeZoneCalculate.getTimeZoneCalculate().getDatetime(), customDevicePairMsg.getCustomerid(), customDevicePairMsg.getPuid(), customDevicePairMsg.getPlatformid());
                    CustomDevicePair.model()
                            .setPairEnd(timeZoneCalculate.getTimeZoneCalculate().getDatetime())
                            .where(CustomDevicePair::getId).eq(queryList.getFirst().getId())
                            .update();
                }else{
                    log.info("found no rows in DB, ignore this msg: {}",
                            receivedTxt);
                }
            }
        }
    }

    private TimeZoneCalculateExt calculateZoneDateTime(String zoneId, long time) {
        TimeZoneCalculateExt result = TimeZoneCalculateExt.builder().build();
        TimeZoneCalculate timeZoneCalculate = TimeZoneCalculate.builder().build();
        boolean flag = false;
        try{
            if(StringUtils.isNotEmpty(zoneId)){
                ZoneId specificZoneId = ZoneId.of(zoneId);
                ZonedDateTime zonedDateTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(time), specificZoneId);
                String offset = !zonedDateTime.getOffset().getId().contains(":") ? "0:00" : zonedDateTime.getOffset().getId();
                String zoneTime = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(zonedDateTime);
                timeZoneCalculate = TimeZoneCalculate.builder().datetime(zoneTime).timezone(Integer.parseInt(offset.split(":")[0])).build();
                flag = true;
            }
        }catch (Exception e){
            log.error("calculateZoneDateTime encounterred errors: {}", e.getMessage());
        }

        if(!flag){
            ZoneId specificZoneId = ZoneId.of(DEFAULT_ZERO_TIMEZONE_ID);
            ZonedDateTime zonedDateTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(time), specificZoneId);
            String zoneTime = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(zonedDateTime);
            timeZoneCalculate = TimeZoneCalculate.builder().datetime(zoneTime).timezone(0).build();
        }

        result.setFlag(flag);
        result.setTimeZoneCalculate(timeZoneCalculate);

        return result;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        workerPool = Schedulers.newBoundedElastic(poolSize, waitQueueLength, "worker-pool");
    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    static class TimeZoneCalculate {
        private String datetime;
        private int timezone;

    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    static class TimeZoneCalculateExt {
        private TimeZoneCalculate timeZoneCalculate;
        private boolean flag;
    }

    @Override
    public void run(String... args) {
        msgConsummer().subscribe(Mono::subscribe);
    }
}